package com.shujia.mr.combine;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.CombineFileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.CombineTextInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class CombineCount {
    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        // 在Main函数中对当前的MapReduce进行做配置工作

        // 1.需要创建一个整体的对象job => MapReduce

        Configuration conf = new Configuration();
        Job job = new Job(conf);
        job.setJobName("idea-CombineCount");
        job.setJarByClass(CombineCount.class);

        // 2.告诉Job对应的Map和Reduce具体使用哪个类
        job.setMapperClass(CombineMapper.class); // Class<? extends Mapper
        job.setReducerClass(CombineReducer.class);

        // 3.告诉Job Map端输出的数据类型 和 Reduce端输出的数据类型(最终输出数据类型)
        // java.lang.ClassCastException: interface javax.xml.soap.Text 类型转换错误
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);


        // 设置一个Combiner预聚合类 该类要求是Reducer的子类
        /**
         * TODO:探究Combiner的作用
         * Map-Reduce Framework
         *
         * MapTask1:
         * 		Map input records=589823
         * 		Map output records=1769469
         * 	    Combine input records=0
         * 	    Spilled Records=1769469
         *
         * MapTask2:
         *      Map input records=557056
         * 		Map output records=1671168
         * 	    Combine input records=0
         *
         * MapTask3:
         *      Map input records=458760
         * 		Map output records=1376280
         *
         *
         * 	ReduceTask:
         * 	   Reduce input records=4816917
         *
         *
         * 	总计：
         * 	    Map output records=4816917
         * 	    Reduce input records=4816917
         *
         *
         * 添加Combiner之后的效果：
         *      Map input records=589823
         * 		Map output records=1769469
         * 		Map output bytes=15925221
         * 		Map output materialized bytes=67
         * 		Input split bytes=239
         * 		Combine input records=1769469
         * 		Combine output records=5
         *
         * 	MapTask2:
         * 	    Map input records=557056
         * 		Map output records=1671168
         * 		Map output bytes=15040512
         * 		Map output materialized bytes=67
         * 		Input split bytes=239
         * 		Combine input records=1671168
         * 		Combine output records=5
         *
         * MapTask3:
         *      Map input records=458760
         * 		Map output records=1376280
         * 		Map output bytes=12386520
         * 		Map output materialized bytes=67
         * 		Input split bytes=320
         * 		Combine input records=1376280
         * 		Combine output records=5
         *
         * 	ReduceTask:
         * 	    Reduce input groups=5
         * 		Reduce shuffle bytes=183
         * 		Reduce input records=15
         * 		Reduce output records=5
         *
         * 	   输入数据是15条 最终输出是5条
         *
         *     预聚合:表示提前进行做聚合操作
         */


        job.setCombinerClass(CombineReducer.class);

        job.setInputFormatClass(CombineTextInputFormat.class); // 问题
        CombineFileInputFormat.setMaxInputSplitSize(job,5 * 1024 * 1024L);


        // 设置ReduceTask数量为2  => map Task 3 Reduce Task 2
        job.setNumReduceTasks(2);



        // 4.需要告诉MapReduce数据所在位置  输入数据位置  输出数据位置
        //   路径选择时需要注意：如果在IDEA中执行，那么路径是读取当前本地路径
//                              如果是在Hadoop集群中执行，那么需要设置HDFS中的路径
        FileInputFormat.addInputPath(job,new Path("hadoopCode/data/mapreduce/word"));
        FileOutputFormat.setOutputPath(job,new Path("hadoopCode/output/combiner"));

//        FileInputFormat.addInputPath(job,new Path("/input"));
//        FileOutputFormat.setOutputPath(job,new Path("/mapreduce/output"));



        // 5.做代码提交
        //  在本地IDEA中执行，需要有Hadoop环境，并且能获取到winUtils
        //  Process finished with exit code 0  退出代码为0 并不代表当前任务执行成功
        job.waitForCompletion(true);

    }
}
